package club.vann.kafka.mock.service;

import club.vann.kafka.mock.kafka.KafkaProducer;
import club.vann.kafka.mock.util.AcceptorUtil;
import com.cyberx.lunyu.AcceptorOuterClass;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import lombok.RequiredArgsConstructor;
import org.springframework.stereotype.Service;
import org.springframework.util.ObjectUtils;

@RequiredArgsConstructor
@Service
public class QuoteAcceptorService {
  private static final Map<String, Map<String, String>> QUOTE_INFO_MAP = new ConcurrentHashMap<>();
  private final KafkaProducer kafkaProducer;

  public void request(AcceptorOuterClass.Acceptor.AcceptorRequest request) {
    long now = System.currentTimeMillis();
    String clientRequestId = request.getClientRequestId();
    String responseTopic = request.getTopic();
    String id = UUID.randomUUID().toString();
    Map<String, String> dataMap = request.getDataMap();
    Map<String, String> responseMap = new HashMap<>();
    responseMap.put("ID", id);
    responseMap.put("PRICE", "200");
    responseMap.put("EXPIRED_TIMESTAMP", now + 100000 + "");
    responseMap.putAll(dataMap);
    QUOTE_INFO_MAP.put(id, responseMap);
    AcceptorOuterClass.Acceptor.AcceptorResponse acceptorResponse =
        AcceptorUtil.ofResponse(
            clientRequestId,
            AcceptorOuterClass.Acceptor.Status.COMPLETED,
            responseTopic,
            "",
            responseMap);
    this.kafkaProducer.send(responseTopic, clientRequestId, acceptorResponse.toByteArray());
  }

  public void confirm(AcceptorOuterClass.Acceptor.AcceptorRequest request) {
    String clientRequestId = request.getClientRequestId();
    String responseTopic = request.getTopic();
    Map<String, String> dataMap = request.getDataMap();
    String id = dataMap.get("ID");
    if (ObjectUtils.isEmpty(QUOTE_INFO_MAP.get(id))) {
      return;
    }
    Map<String, String> requestMap = QUOTE_INFO_MAP.get(id);

    Map<String, String> responseMap = new HashMap<>();
    responseMap.put("TICKER", requestMap.get("TICKER"));
    responseMap.put("STATE", "COMPLETED");
    responseMap.put("SIDE", requestMap.get("SIDE"));
    responseMap.put("AVG_FILLED_PRICE", "65200.67");
    responseMap.put("FILLED_QTY", "0.2");
    responseMap.put("ID", requestMap.get("ID"));
    responseMap.put("TOTAL_FEE", "20");
    responseMap.put("CLIENT_ID", requestMap.get("CLIENT_ID"));

    AcceptorOuterClass.Acceptor.AcceptorResponse acceptorResponse =
        AcceptorUtil.ofResponse(
            clientRequestId,
            AcceptorOuterClass.Acceptor.Status.COMPLETED,
            responseTopic,
            "",
            responseMap);
    this.kafkaProducer.send(responseTopic, clientRequestId, acceptorResponse.toByteArray());
  }

  public void cancel(AcceptorOuterClass.Acceptor.AcceptorRequest request) {
    String clientRequestId = request.getClientRequestId();
    String responseTopic = request.getTopic();
    Map<String, String> dataMap = request.getDataMap();
    String id = dataMap.get("ID");
    if (ObjectUtils.isEmpty(QUOTE_INFO_MAP.get(id))) {
      return;
    }
    Map<String, String> requestMap = QUOTE_INFO_MAP.get(id);

    Map<String, String> responseMap = new HashMap<>();
    responseMap.put("TICKER", requestMap.get("TICKER"));
    responseMap.put("STATE", "COMPLETED");
    responseMap.put("SIDE", requestMap.get("SIDE"));
    responseMap.put("AVG_FILLED_PRICE", "65200.67");
    responseMap.put("FILLED_QTY", "0.2");
    responseMap.put("ID", requestMap.get("ID"));
    responseMap.put("TOTAL_FEE", "200");
    responseMap.put("CLIENT_ID", requestMap.get("CLIENT_ID"));

    AcceptorOuterClass.Acceptor.AcceptorResponse acceptorResponse =
        AcceptorUtil.ofResponse(
            clientRequestId,
            AcceptorOuterClass.Acceptor.Status.COMPLETED,
            responseTopic,
            "",
            responseMap);
    this.kafkaProducer.send(responseTopic, clientRequestId, acceptorResponse.toByteArray());
  }
}
